package one.bugu.zookeeper.rpc.framework.service.socket;

import one.bugu.zookeeper.rpc.framework.entity.Message;
import one.bugu.zookeeper.rpc.framework.service.RpcServiceConfiguration;
import one.bugu.zookeeper.rpc.framework.service.handler.ProcessHandler;
import com.google.gson.Gson;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created with IntelliJ IDEA.
 * User: LangK
 * Created Date 2019/2/11
 * Time: 16:42
 * Description:
 */
public class ServiceSocket implements Runnable {

    private Gson gson = new Gson();
    private static final int MAX_LENGTH = 1024;
    private Logger logger = LoggerFactory.getLogger(ServiceSocket.class);

    private RpcServiceConfiguration rpcServiceConfiguration;

    public ServiceSocket(RpcServiceConfiguration rpcServiceConfiguration) {
        this.rpcServiceConfiguration = rpcServiceConfiguration;
    }

    public void run() {
        NioEventLoopGroup boos = new NioEventLoopGroup(8);
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            ChannelFuture future = bootstrap
                    .group(boos, worker)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<NioSocketChannel>() {
                        protected void initChannel(NioSocketChannel ch) {
                            ch.pipeline().addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
                            ch.pipeline().addLast("decoder", new StringDecoder());
                            ch.pipeline().addLast("encoder", new StringEncoder());
                            ch.pipeline().addLast("handler", new ServerHandler());
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, MAX_LENGTH)
                    //维持链接的活跃，清除死链接(SocketChannel的设置)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    //关闭延迟发送
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .bind(Integer.parseInt(rpcServiceConfiguration.getPort())).sync();
            //等待服务端监听端口关闭
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            logger.error("RPC长链接服务端异常", e);
        } finally {
            worker.shutdownGracefully().syncUninterruptibly();
            boos.shutdownGracefully().syncUninterruptibly();
        }

    }


    class ServerHandler extends SimpleChannelInboundHandler<String> {

        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msgStr) throws Exception {
            try {
                logger.info("接收到RPC请求：{}", msgStr);
                Message message = gson.fromJson(msgStr, Message.class);
                String object = ProcessHandler.process(message.getMessage());
                message.setResult(object);
                channelHandlerContext.writeAndFlush(gson.toJson(message) + "\r\n");
                return;
            } catch (Exception e) {
                logger.error("RPC服务端处理异常", e);
                ReferenceCountUtil.release(msgStr);
            }
        }

        /*
         *
         * 覆盖 channelActive 方法 在channel被启用的时候触发 (在建立连接的时候)
         *
         * channelActive 和 channelInActive 在后面的内容中讲述，这里先不做详细的描述
         * */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            logger.info("RamoteAddress : " + ctx.channel().remoteAddress() + " active !");
            super.channelActive(ctx);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }
}
